package com.afdor.rws.event.driven.support;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.driven.EventChangeListener;
import com.afdor.rws.event.driven.EventChangeManager;
import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.entities.Event;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

/**
 * 事件状态变更管理器
 *
 * @author 悭梵
 * @date Created in 2018-12-05 18:05
 */
@Slf4j
@Extension(value = "default", order = Extension.NORMAL_ORDER - 100, interfaceClass = EventChangeManager.class)
public class DefaultEventChangeManager implements EventChangeManager, Runnable {

    protected AtomicBoolean runing = new AtomicBoolean();

    protected volatile List<EventChangeListener> listeners;

    protected final ConcurrentLinkedQueue<Event> eventChangedQueue = new ConcurrentLinkedQueue<>();

    @Override
    public <T extends EventChangeManager> T init() {
        runing.compareAndSet(Boolean.FALSE, Boolean.TRUE);
        // scheduleAtFixedRate  在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务。注意，period是从任务开始执行算起的。开始执行任务后，定时器每隔period时长检查该任务是否完成，如果完成则再次启动任务，否则等该任务结束后才再次启动任务
        // scheduleWithFixedDelay 在initialDelay时长后第一次执行任务，以后每当任务执行完成后，等待delay时长，再次执行任务。
        // 两个方法均在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务，但scheduleAtFixedRate以开始时间计算间隔，而scheduleWithFixedDelay以结束时间计算间隔
        ThreadUtils.getScheduledThreadPool().scheduleWithFixedDelay(this, 0, 100, TimeUnit.MILLISECONDS);
        return (T) this;
    }

    @Override
    public void destroy() {
        runing.compareAndSet(Boolean.TRUE, Boolean.FALSE);
        ThreadUtils.getScheduledThreadPool().remove(this);
    }

    @Override
    public void run() {
        if (!Thread.currentThread().isInterrupted()
                && listeners != null && !listeners.isEmpty()
                && !eventChangedQueue.isEmpty()) {
            /* 线程未终止，就执行事件变更处理业务*/
            while (!eventChangedQueue.isEmpty()) {
                if (Thread.currentThread().isInterrupted()) {
                    break;
                }
                final Event event = eventChangedQueue.poll();
                if (event != null) {
                    listeners.parallelStream()
                            .forEach(listener -> listener.onChange(event));
                }
            }
        }

    }

    @Override
    public synchronized <T extends EventChangeManager> T addEventChangeListener(@NonNull EventChangeListener listener) {
        if (this.listeners == null) {
            this.listeners = new ArrayList<>(10);
        }
        this.listeners.add(listener);
        return (T) this;
    }

    /**
     * 修改事件状态
     *
     * @param event
     */
    @Override
    public void eventChanged(@NonNull final Event event) {
        if (!runing.get()) {
            log.warn("事件状态变更管理器已停止运行，不能再继续执行状态变更相关操作：{}", event);
            return;
        }
        int i = 0;
        while (!eventChangedQueue.offer(event)) {
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(++i * 10));
        }
    }

}
